//package com.culture.bootdemo.mq.rocketmq.customer;
//
//import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
//import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
//import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
//import com.alibaba.rocketmq.client.exception.MQClientException;
//import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
//import com.alibaba.rocketmq.common.message.MessageConst;
//import com.alibaba.rocketmq.common.message.MessageExt;
//import com.culture.bootdemo.mq.rocketmq.customer.MessageHandler;
//import lombok.extern.slf4j.Slf4j;
//import org.apache.commons.lang3.time.DateFormatUtils;
//
//import java.lang.management.ManagementFactory;
//
//@Slf4j
//public class SimpleConsumer {
//
//    private DefaultMQPushConsumer consumer;
//
//    public SimpleConsumer(String namesrvAddr, String consumerGroup) {
//        this.consumer = new DefaultMQPushConsumer();
//        // 实例名称：c@进程id@hostname
//        String pidHostname = ManagementFactory.getRuntimeMXBean().getName();
//        this.consumer.setInstanceName("c@" + pidHostname);
//        this.consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
//        this.consumer.setNamesrvAddr(namesrvAddr);
//        this.consumer.setConsumerGroup(consumerGroup);
//    }
//
//
//    private int retryTime = 3;
//    public int getRetryTime() {
//        return retryTime;
//    }
//
//    public void setRetryTime(int retryTime) {
//        this.retryTime = retryTime;
//    }
//
//
//    public void subscribe(String topic, String tags, MessageHandler messageHandler) throws MQClientException {
//        try {
//            //消费所有的
//            consumer.subscribe(topic, tags);
//            //并发消费
//            consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
//                boolean batchHandle = true;
//                for (MessageExt msg : msgs) {
//                    String message = "";
//                    boolean handle = false;
//                    int reconsumeTimes = 0;
//                    try {
//                        message = new String(msg.getBody(), "utf-8");
//                        String msgId = msg.getMsgId();
//                        long queueOffset = msg.getQueueOffset();
//                        reconsumeTimes = msg.getReconsumeTimes();
//                        long maxOffSet = Long.parseLong(msg.getProperty(MessageConst.PROPERTY_MAX_OFFSET));
//                        log.info("收到消息[msgId={},queueOffset={},reconsumeTimes={},Timestamp={},ESSAGE_ID={},TAGS={},KEYS={},MAX_OFFSET={},DELAY={} msg={} topic={} }]",
//                                msgId,
//                                queueOffset,
//                                reconsumeTimes,
//                                DateFormatUtils.format(
//                                        msg.getBornTimestamp(),
//                                        "yyyy-MM-dd HH:mm:ss"),
//                                msg.getProperty(MessageConst.PROPERTY_ORIGIN_MESSAGE_ID),
//                                msg.getProperty(MessageConst.PROPERTY_TAGS),
//                                msg.getProperty(MessageConst.PROPERTY_KEYS),
//                                maxOffSet,
//                                msg.getProperty(MessageConst.PROPERTY_DELAY_TIME_LEVEL), message, topic);
//
//                        handle = messageHandler.handle(message);
//                        if (reconsumeTimes >= retryTime) {
//                            handle = true;
//                        }
//                    } catch (Exception e) {
//                        if (reconsumeTimes >= retryTime) {
//                            handle = true;
//                        } else {
//                            handle = false;
//                        }
//                    }
//                    //错误后继续下一个元素继续消费。集体重试
//                    if (!handle) {
//                        batchHandle = false;
//                    }
//                }
//                return batchHandle ? ConsumeConcurrentlyStatus.CONSUME_SUCCESS : ConsumeConcurrentlyStatus.RECONSUME_LATER;
//            });
//            consumer.start();
//        } catch (Exception e) {
//            log.error("消费者[" + consumer.getConsumerGroup() + "][" + consumer.getInstanceName() + "]订阅失败", e);
//            throw e;
//        }
//    }
//
//
//    public void start() {
//        if (consumer != null) {
//            try {
//                consumer.start();
//                log.info("消费者[{}][{}]启动成功", consumer.getConsumerGroup(), consumer.getInstanceName());
//            } catch (Exception e) {
//                log.error("消费者[" + consumer.getConsumerGroup() + "]["
//                        + consumer.getInstanceName() + "]启动失败", e);
//            }
//        }
//    }
//
//}
